package cn.com.klun.tjy.other.zkUsed;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;

import static org.apache.zookeeper.ZooDefs.Ids.OPEN_ACL_UNSAFE;

public class Master implements Watcher{
    static ZooKeeper zk;
    String hostPort;
    static Random random = new Random();//生成id
    static String serverId  = Integer.toHexString(random.nextInt());
    //static boolean isLeader = false;
    private String state = "";// 状态信息
    ChildrenCache workersCache = ChildrenCache.getInstance(); // 该变量用于保存上次获得的从节点列表的本地缓存
    private Logger logger = LoggerFactory.getLogger(Master.class);

    // 检测主节点
    void checkMaster(){
        zk.getData("/master",false,masterCheckCallBack,null);
    }
    //检查是否主节点-回调函数
    AsyncCallback.DataCallback masterCheckCallBack = new AsyncCallback.DataCallback() {
        @Override
        public void processResult(int i, String s, Object o, byte[] bytes, Stat stat) {
            switch (KeeperException.Code.get(i)){
                case CONNECTIONLOSS:
                    checkMaster();
                    return;
                case NONODE:
                    runForMaster();
                    return;
            }
        }
    };

    public Master(String hostPort) {
        this.hostPort = hostPort;
        try { //创建master的时候初始化worker
            if(workersCache.getCache() == null ){
                workersCache.setCache(zk.getChildren("/workers",false));
            }
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void startZk() throws IOException {
        zk = new ZooKeeper(hostPort,15000,this);
    }
    public void stopZk() throws InterruptedException {
        zk.close();
    }

    @Override
    public void process(WatchedEvent watchedEvent) {
        System.out.println("****************"+watchedEvent);
    }

    void runForMaster(){
        zk.create("/master",serverId.getBytes(),OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL,masterCreateCallback,null);
    }

    /**
     * 行使主节点的主权
     */
    public void takeLeadership(){
        logger.info("##################我在行使领导的权利:(主权拥有着的ID："+serverId+"");
        getTasks();
    }

    /**
     * 创建主节点回调函数
     * @param args
     * @throws IOException
     * @throws InterruptedException
     */
    AsyncCallback.StringCallback masterCreateCallback = new AsyncCallback.StringCallback() {
        @Override
        public void processResult(int i, String s, Object o, String s1) {
            switch (KeeperException.Code.get(i)) {
                case CONNECTIONLOSS:
                    logger.info("************创建主节点失去连接，检测主节点数据");
                    checkMaster();
                    break;
                case OK:
                    //isLeader = true;//返回的是主节点
                    /*state = MasterStates.ELECTED;*/
                    logger.info("************创建主节点成功，行使主节点权力");
                    state = "ELECTED";
                    takeLeadership(); //如果返回OK，那么开始行使领导权
                    break;
                case NODEEXISTS:
                    /*state = MasterStates.NOTELECTED;*/
                    logger.info("************主节点已经存在，继续检测主节点");
                    state = "NOTELECTED";
                    masterExists();
                    break;
                default:
                    //state = MasterStates.NOTELECTED;
                    state = "NOTELECTED";
                    //isLeader = false;运行错误
                    logger.error("运行错误",KeeperException.create(KeeperException.Code.get(i),s));
            }
        }
    };
    // 如果存在此注解点设置一个监视节点
    void masterExists(){
        zk.exists("/master",masterExistsWatcher,masterExistsCallback,null);
    }
    Watcher masterExistsWatcher = new Watcher() {
        @Override
        public void process(WatchedEvent watchedEvent) {
            if(watchedEvent.getType() == Event.EventType.NodeDeleted){
                //logger.info("");
                assert "/master".equals(watchedEvent.getPath());
            }
        }
    };
    // 主节点存在的回调函数
    AsyncCallback.StatCallback masterExistsCallback = new AsyncCallback.StatCallback() {
        @Override
        public void processResult(int i, String s, Object o, Stat stat) {
            switch (KeeperException.Code.get(i)){
                case CONNECTIONLOSS:
                    masterExists(); //丢失链接设置监视点？
                    break;
                case OK:
                    if(null == state){
                        /*state = MasterStates.RUNING;*/
                        state = "RUNING";
                        runForMaster();//尝试注册主节点
                    }
                    break;
                default:
                    checkMaster(); //检测主节点，如果不存在则再注册主节点
                    break;
            }
        }
    };
    // /workers 中子节点变化的 监视节点
    Watcher workersChangeWatcher = new Watcher() {
        @Override
        public void process(WatchedEvent watchedEvent) {
            if(watchedEvent.getType() == Event.EventType.NodeChildrenChanged){
                assert "/workers".equals(watchedEvent.getPath());
                getWorkers();
            }
        }
    };

    // 获取工作节点 -whm
    void getWorkers(){
        zk.getChildren("/workers",workersChangeWatcher,workersGetChildrenCallback,null);
    }
    AsyncCallback.ChildrenCallback workersGetChildrenCallback = new AsyncCallback.ChildrenCallback() {
        @Override
        public void processResult(int i, String s, Object o, List<String> list) {
            switch (KeeperException.Code.get(i)){
                case CONNECTIONLOSS:
                    getWorkers();
                    break;
                case OK:
                    logger.info("成功获取到workers"+list.size()+"workers");
                    reassignAndSet(list);//重新分配崩溃的从节点的任务，并设置行的从节点列表
                    break;

                default:
                    logger.error("运行错误");
            }
        }
    };
    //重新分配崩溃的从节点的任务，并设置行的从节点列表
    void reassignAndSet(List<String> children){
        List<String> toProcess;
        // 如果本身为空直接赋值
        if(workersCache.getCache().size() == 0){
            workersCache.setCache(children);
            toProcess = null;
        }else{
            logger.info("清除配置");
            // 如果不是第一次，那么检查是否有从节点已经被移除了
            toProcess = removeAndSet(children);
        }
        if(toProcess != null){ // toProcess
            for(String worker : toProcess){
                getAbsentWorkerTasks(worker);//如果从节点被移除了 需要重新分配任务
            }
        }
    }
    // 筛选出被移除的节点 + 更新最新的节点
    List<String> removeAndSet(List<String> children){
        List<String> toProcess = new ArrayList<>();
        List<String> cache = workersCache.getCache();
        for (String child:cache) {
            if(!children.contains(child)){ //新删除的节点
                toProcess.add(child);
            }
        }
        //最后新赋值
        workersCache.setCache(children);
        return toProcess;
    }
    void getAbsentWorkerTasks(String worker){
        //如果从节点被移除了 需要重新分配任务
        /*找出被删除的workers对应的任务，并重新分配任务到一个新的worker下 */
        zk.getData("/assgin/"+worker,false,getAssginDateCallback,null);
    }
    AsyncCallback.DataCallback getAssginDateCallback = new AsyncCallback.DataCallback() {
        @Override
        public void processResult(int i, String s, Object o, byte[] bytes, Stat stat) {
            switch (KeeperException.Code.get(i)){
                case CONNECTIONLOSS:
                    String worker = (String)o; //验证是不是和上面的worker相同
                    getAbsentWorkerTasks(worker);
                    break;
                case OK:
                    //重新分配任务给worker
                    int workerNum  = random.nextInt(workersCache.getCache().size());
                    String designateWorker  = workersCache.getCache().get(workerNum);
                    //把任务分配给worker
                    String assginmentPath = "/assign/"+designateWorker+"/"+(String)o;
                    createAssignment(assginmentPath,bytes);
                    break;
                default:
                    logger.error("getAssginDateCallback 出现异常情款");
            }
        }
    };

    /*主节点等待新任务进行分配*/
    Watcher tasksChangeWatcher = new Watcher() {
        @Override
        public void process(WatchedEvent watchedEvent) {
            assert "/tasks".equals(watchedEvent.getPath());
            getTasks();
        }
    };
    // 获取任务 并放置监视点
    void getTasks(){
        zk.getChildren("/tasks",tasksChangeWatcher,taskGetChildrenCallback,null);
    }
    AsyncCallback.ChildrenCallback taskGetChildrenCallback = new AsyncCallback.ChildrenCallback() {
        @Override
        public void processResult(int i, String s, Object o, List<String> list) {
            switch (KeeperException.Code.get(i)){
                case CONNECTIONLOSS:
                    getTasks();
                    break;
                case OK:
                    if(list != null){// children 是 tasks
                        assignTasks(list);
                    }
                    break;
                default:
                    logger.error("getChildren failed");

            }
        }
    };

    // 分配列表中的任务 分配给空闲的 worker
    void assignTasks(List<String> tasks){
        for(String task : tasks){
            logger.info("$$$$$$$$第个任务为："+task);
            getTaskData(task);
        }
    }
    void getTaskData(String task){
        zk.getData("/tasks/"+task,false,taskDataCallback,task);
    }

    AsyncCallback.DataCallback taskDataCallback = new AsyncCallback.DataCallback() {
        @Override
        public void processResult(int i, String s, Object o, byte[] data, Stat stat) {
            switch (KeeperException.Code.get(i)){
                case CONNECTIONLOSS:
                    getTaskData((String)o);
                    break;
                case OK:
                    // 不需要 重新更新workerlist 再分配任务 设置监视点 一有变动就触发更新，并设置新的监视点
                    // 随机获取 worker
                    int worker  = random.nextInt(workersCache.getCache().size());
                    String designateWorker  = workersCache.getCache().get(worker);
                    // 把任务分配给 worker
                    String assginmentPath = "/assign/"+designateWorker+"/"+(String)o;
                    createAssignment(assginmentPath,data);
                    for (int j = 0; j < data.length; j++) {
                        System.out.println("*********获取的数据"+data[j]);
                    }
                    break;
                default:
                    logger.error("*******获取任务数据异常;:::"+i);
            }

        }
    };
    //创建任务节点
    void createAssignment(String path,byte[] data){
        zk.create(path,data, ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT,assignTasksCallback,data);
    }
    AsyncCallback.StringCallback assignTasksCallback = new AsyncCallback.StringCallback() {
        @Override
        public void processResult(int i, String s, Object o, String s1) {
            switch (KeeperException.Code.get(i)){
                case CONNECTIONLOSS:
                    createAssignment(s,(byte[]) o);
                    break;
                case OK:
                    logger.info("正确的分配任务"+s1);
                    break;
                case NODEEXISTS:
                    logger.warn("任务节点已经存在");
                    break;
                default:
                    logger.error("exception");
            }
        }
    };

    public static void main(String[] args) throws IOException, InterruptedException {
        Master m =  new Master("192.168.137.12:2181");
        //Master m =  new Master("127.0.0.1:2181");
        m.startZk();
        m.runForMaster();
        Thread.sleep(2000l);
//        if(isLeader){
//            System.out.println("这是主节点");
//            Thread.sleep(6000000);
//        }else{
//            System.out.println("不是主节点");
//        }

        m.stopZk();
    }


}




